package com.yunhe.abnormal.listener;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.yunhe.abnormal.client.DeviceClient;
import com.yunhe.abnormal.client.MeasHistoryClient;
import com.yunhe.abnormal.client.MeasRealtimeClient;
import com.yunhe.common.constant.*;
import com.yunhe.abnormal.domain.data.AlarmEvent;
import com.yunhe.abnormal.mapper.data.AlarmEventMapper;
import com.yunhe.common.model.DomainList;
import com.yunhe.common.model.base.AbnormalStatusDTO;
import com.yunhe.common.model.data.AlarmEventDTO;
import com.yunhe.common.model.data.Measurement;
import com.yunhe.common.model.data.MeasurementReqDTO;
import com.yunhe.common.model.device.Analog;
import com.yunhe.common.model.device.EnergyUnitDTO;
import com.yunhe.common.model.model.AnalogRequestBody;
import com.yunhe.common.model.msg.AlarmEventMsgDTO;
import com.yunhe.abnormal.service.base.AbnormalStatusService;
import com.yunhe.abnormal.service.data.AlarmEventService;
import com.yunhe.common.util.JsonUtil;
import com.yunhe.common.util.StringUtil;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

import java.util.*;
import java.util.concurrent.CopyOnWriteArraySet;

/**
 * 健康状态计算监听类
 * @author liuronglei
 */
@Component
public class HealthStatusListener {
    private static final Logger logger = LoggerFactory.getLogger(HealthStatusListener.class);
    private ObjectMapper mapper = JsonUtil.getObjectMapper();

    @Autowired
    private AlarmEventService alarmEventService;
    @Autowired
    private DeviceClient deviceClient;
    @Autowired
    private MeasRealtimeClient measRealtimeClient;
    @Autowired
    private MeasHistoryClient measHistoryClient;
    @Autowired
    private AbnormalStatusService abnormalStatusService;
    @Autowired
    private AlarmEventMapper alarmEventMapper;

    // 保存已提示日志的设备
    private Set<Long> warnDeviceSet = new CopyOnWriteArraySet<>();

    /**
     * 获得需要展示的健康状态ID
     */
    private List<Long> getAbnormalStatusIds() {
        List<Long> abnormalStatusIds = new ArrayList<>();
        AbnormalStatusDTO statusReady = abnormalStatusService.findByName(AbnormalConsts.ABNORMAL_STATUS.READY.getName());
        if (statusReady != null) {
            abnormalStatusIds.add(statusReady.getId());
        }
        AbnormalStatusDTO sentReady = abnormalStatusService.findByName(AbnormalConsts.ABNORMAL_STATUS.SENT.getName());
        if (sentReady != null) {
            abnormalStatusIds.add(sentReady.getId());
        }
        return abnormalStatusIds;
    }

    /**
     * 获得设备对应点号Map
     * deviceId -> (typeName -> analog)
     * @param deviceIds 设备ID列表
     */
    private Map<Long, Map<String, Analog>> getDeviceAnalogMap(List<Long> deviceIds, List<String> typeNames) {
        Map<Long, Map<String, Analog>> result = new HashMap<>();
        AnalogRequestBody requestBody = new AnalogRequestBody();
        requestBody.setDeviceId(deviceIds);
        List<Analog> analogs = deviceClient.getAnalogsByDevAndTypeForPost(typeNames, requestBody).getBody().getResults();
        if (analogs!=null && !analogs.isEmpty()) {
            for (Analog analog : analogs) {
                String typeName = analog.getTypeName();
                Long deviceId = analog.getDeviceId();
                Map<String, Analog> deviceAnalogMap = result.computeIfAbsent(deviceId, v -> new HashMap<>());
                deviceAnalogMap.put(typeName, analog);
            }
        }
        return result;
    }

    /**
     * 根据设备ID和点号数据类型获得点号
     * @param deviceAnalogMap 设备和点号Map
     * @param deviceId 设备ID
     * @param typeName 能量单元类型列表
     */
    private Long getPointNumber(Map<Long, Map<String, Analog>> deviceAnalogMap, Long deviceId, String typeName) {
        Map<String, Analog> storageAnalogMap = deviceAnalogMap.get(deviceId);
        if (storageAnalogMap != null) {
            Analog analog = storageAnalogMap.get(typeName);
            if (analog != null) {
                return analog.getPointNumber();
            }
        }
        return null;
    }

    /**
     * 获得储能和新能源发电单元的点号列表
     * @param deviceAnalogMap 设备和点号Map
     * @param energyUnits 能量单元ID列表
     */
    private List<Long> getAnalogs(Map<Long, Map<String, Analog>> deviceAnalogMap,
                                                   List<Long> energyUnits) {
        List<Long> result = new ArrayList<>();
        for (Long energyUnitId : energyUnits) {
            // 健康状态
            Long workStatusPoint = getPointNumber(deviceAnalogMap, energyUnitId, DeviceConsts.POINT_DATA_TYPE.HEALTH_STATUS.value());
            if (workStatusPoint != null) {
                result.add(workStatusPoint);
            }
        }
        return result;
    }

    /**
     * 保存健康状态
     * @param result 点号值Map
     */
    private void saveAlarmStatus(Map<Long, Double> result) {
        List<Measurement> list = new ArrayList<>();
        result.forEach((k, v) -> {
            Measurement measurement = new Measurement();
            measurement.setPointNumber(k);
            measurement.setRtime(System.currentTimeMillis());
            measurement.setVal(v);
            list.add(measurement);
        });
        DomainList<Measurement> domainList = new DomainList<>(list);
        measRealtimeClient.saveMeasureRealtime(domainList);
        measHistoryClient.createMeasurementAsync(domainList, MeasurementConsts.FREQUENCY.Original.value());
    }

    /**
     * 监听告警内容刷新消息
     * @param record 消息内容：主要是电站ID
     */
    @KafkaListener(topics = KafkaConsts.Topic.AlarmEventRefresh, groupId = KafkaConsts.Group.HealthStatus)
    public void alarmStatusListener(ConsumerRecord<?, String> record) {
        try {
            String value = record.value();
            if (!StringUtil.isEmpty(value)) {
                AlarmEventMsgDTO msg = mapper.readValue(value, AlarmEventMsgDTO.class);
                String action = msg.getAction();
                List<AlarmEventDTO> alarmEvents = msg.getAlarmEvents();
                // 因为目前告警并没有保存能量单元ID，所有只要电站下的告警有变动，就更新电站下的所有能量单元的健康状态
                // 判断哪些电站健康状态需要刷新
                // 组建本次消息电站和告警的映射
                Map<Long, List<AlarmEventDTO>> stationCurrentMap = new HashMap<>();
                alarmEvents.forEach(item -> {
                    Long stationId = item.getStationId();
                    stationCurrentMap.computeIfAbsent(stationId, v -> new ArrayList<>()).add(item);
                });
                // 获得这些电站所有告警
                List<Long> abnormalStatusIds = getAbnormalStatusIds();
                Map<Long, List<AlarmEventDTO>> stationMaps = new HashMap<>();
                Iterable<AlarmEvent> stationAlarmEvents = alarmEventService.findByCondition(null, new ArrayList<>(stationCurrentMap.keySet()),
                        null, null, null, abnormalStatusIds, null, null, null, null);
                List<AlarmEventDTO> stationAlarmEventDTOs = alarmEventMapper.to((List<AlarmEvent>)stationAlarmEvents);
                if (stationAlarmEventDTOs !=null && !stationAlarmEventDTOs.isEmpty()) {
                    stationAlarmEventDTOs.forEach(item -> {
                        Long stationId = item.getStationId();
                        stationMaps.computeIfAbsent(stationId, v -> new ArrayList<>()).add(item);
                    });
                }
                // 循环这些告警，组建能量单元和最高健康状态的映射
                Map<Long, Double> energyUnitValueMap = new HashMap<>();
                stationCurrentMap.forEach((stationId, currentAlarms) -> {
                    // 添加最新有告警的能量单元的健康状态
                    List<AlarmEventDTO> alarmEventDTOs = stationMaps.get(stationId);
                    if (alarmEventDTOs!=null && !alarmEventDTOs.isEmpty()) {
                        alarmEventDTOs.forEach(item -> {
                            // 找不到告警等级或者告警等级为轻微时，跳过
                            AbnormalConsts.ALARM_LEVEL alarmLevel = AbnormalConsts.ALARM_LEVEL.byValue(item.getAlarmLevelName());
                            if (alarmLevel != null && !AbnormalConsts.ALARM_LEVEL.IGNORE.equals(alarmLevel)) {
                                Long deviceId = item.getDevId();
                                Long energyUnitId = deviceId;
                                EnergyUnitDTO energyUnit = deviceClient.getEnergyUnitByDeviceId(deviceId).getBody().getResults();
                                if (energyUnit != null) {
                                    energyUnitId = energyUnit.getId();
                                }
                                Double topAlarmLevel = energyUnitValueMap.get(energyUnitId);
                                if (topAlarmLevel == null || alarmLevel.code() > topAlarmLevel) {
                                    energyUnitValueMap.put(energyUnitId, alarmLevel.code().doubleValue());
                                }
                            }
                        });
                    }
                    // 判断哪些能量单元告警消除
                    // 添加情况下，肯定不会出现从不健康变健康的情况，所以跳过添加的情况
                    if (!SystemConsts.ACTION.ADD.equals(action)) {
                        currentAlarms.forEach(item -> {
                            Long deviceId = item.getDevId();
                            Long energyUnitId = deviceId;
                            EnergyUnitDTO energyUnit = deviceClient.getEnergyUnitByDeviceId(deviceId).getBody().getResults();
                            if (energyUnit != null) {
                                energyUnitId = energyUnit.getId();
                            }
                            // 如果本次消息中有告警的设备，但最新的告警中没有，则需要将健康状态置为正常
                            energyUnitValueMap.computeIfAbsent(energyUnitId, v -> AbnormalConsts.ALARM_LEVEL.NORMAL.code().doubleValue());
                        });
                    }
                });

                Map<Long, Double> result = new HashMap<>();
                if (energyUnitValueMap.size() > 0) {
                    List<Long> energyUnitIds = new ArrayList<>(energyUnitValueMap.keySet());
                    // 获得设备ID对应的点号Map
                    Map<Long, Map<String, Analog>> deviceAnalogMap = getDeviceAnalogMap(energyUnitIds,
                            Collections.singletonList(DeviceConsts.POINT_DATA_TYPE.HEALTH_STATUS.value()));
                    List<Long> analogs = getAnalogs(deviceAnalogMap, energyUnitIds);
                    if (analogs != null && !analogs.isEmpty()) {
                        // 获得所有点号值
                        MeasurementReqDTO reqDTO = new MeasurementReqDTO();
                        reqDTO.setPointNumber(analogs);
                        Map<Long, Double> analogValueMap = measRealtimeClient.getMeasureRealtime(reqDTO).getBody().getResults();
                        if (deviceAnalogMap.size() > 0) {
                            energyUnitValueMap.forEach((energyUnitId, status) -> {
                                Long alarmStatusPoint = getPointNumber(deviceAnalogMap, energyUnitId, DeviceConsts.POINT_DATA_TYPE.HEALTH_STATUS.value());
                                if (alarmStatusPoint == null) {
                                    if (!warnDeviceSet.contains(energyUnitId)) {
                                        warnDeviceSet.add(energyUnitId);
                                        logger.warn("健康状态计算被跳过：找不到健康状态点号，能量单元{}", energyUnitId);
                                    }
                                } else {
                                    if (analogValueMap != null && !status.equals(analogValueMap.get(alarmStatusPoint))) {
                                        result.put(alarmStatusPoint, status);
                                    }
                                }
                            });
                        }
                    }
                }
                // 对结果进行保存
                if (result.size() > 0) {
                    saveAlarmStatus(result);
                }
            }
        } catch (Exception e) {
            logger.error("健康状态计算报错，{}", record.value());
            e.printStackTrace();
        }
    }
}